package cn.ac.iie.di.ban.data.exchange.runner.slave.RocketMQHelper.receiveMQ;

import cn.ac.iie.di.datadock.rdata.exchange.client.core.session.receive.REAbstractReceiveMessageHandler;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ArrayBlockingQueue;

public class MessageSuccessReceiver extends REAbstractReceiveMessageHandler<REMessage> {

    private static final Logger LOG = LoggerFactory.getLogger(MessageSuccessReceiver.class);
    private ArrayBlockingQueue<REMessage> queue;

    public MessageSuccessReceiver(ArrayBlockingQueue<REMessage> queue) {
        this.queue = queue;
    }


    @Override
    public boolean handle(REMessage message) {
        if (message != null) {
            LOG.info("get 1 message, message size: " + message.data.size());
            putMessage(message);
        }
        return true;
    }

    private void putMessage(REMessage message) {
        try {
            this.queue.put(message);
        } catch (InterruptedException e) {
        }
    }
}
